package com.roy.kfk.basic;

import org.apache.kafka.clients.producer.*;

import java.util.Map;
import java.util.Properties;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;

/**
 * @auth roykingw
 */
public class MyProducer {
    private static final String BOOTSTRAP_SERVERS = "worker1:9092,worker2:9092,worker3:9092";
    private static final String TOPIC = "disTopic";

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        //1.配置发送者相关数据，包括kafka地址，key-value 序列化方式
        Properties props = new Properties();
        // 1.1 此处配置的是kafka 服务器地址
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
        // 1.2 这里 配置了我们自定义的消息拦截机制 【***】
        props.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG,"com.roy.kfk.basic.MyInterceptor");
        /**
         * 1. 下面我们为什么需要设置 kafka key-value 消息的序列化方式呢？
         *      - ⽣产者如何将消息的key和value序列化成⼆进制数据。
         */
        // 1.3 配置key的序列化类
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");
        // 1.4 配置value的序列化类
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");

        // 2. 下面构建消息体内容
        Producer<String,String> producer = new KafkaProducer<>(props);
//        CountDownLatch latch = new CountDownLatch(5);
        for(int i = 0; i < 10; i++) {
            //2.1 构建消息
            // key = i ,value 消息的内容就是  "MyProducer" + i
            // key是⽤来进⾏分区的可选项。Kafka通过key来判断消息要分发到哪个Partition。
            // 如果没有填写key，那么Kafka会⾃动选择Partition。
            // 如果填写了key，那么会通过声明的Serializer序列化接⼝，将key转换成⼀个byte[]数组，然后对key进⾏hash，选择Partition。这样可以保证key相同的
            // 消息会分配到相同的Partition中。
            ProducerRecord<String, String> record = new ProducerRecord<>(TOPIC, Integer.toString(i), "MyProducer" + i);
            //2.2 发送消息
            // 2.2.1 不接受服务端响应 的发送
            //单向发送：不关心服务端的应答。
            // producer.send(record); // 这种方式发送的消息，不会接受服务器应答就返回，因为没有确定是否收到服务器应答，所以这种方式可能造成数据丢失
            // System.out.println("message "+i+" sended");

            // 2.2.2 发送消息之后，客户端需要接受服务器响应 才能继续往下执行
            // 通过这个 响应结果 RecordMetadata，我们班可以知道服务端，有没有把那条消息处理成功
            //同步发送：获取服务端应答消息前，会阻塞当前线程。
            RecordMetadata recordMetadata = producer.send(record).get();
            String topic = recordMetadata.topic(); // 当前消息发送到的 topic 是哪一个
            int partition = recordMetadata.partition();// 当前消息发送到当前topic中的哪个分区
            long offset = recordMetadata.offset();// 消息的偏移量
            String message = recordMetadata.toString();//消息的内容
            System.out.println("message:["+ message+"] sended with topic:"+topic+"; partition:"+partition+ ";offset:"+offset);

            // 2.2.3 消息的异步发送
            //异步发送：消息发送后不阻塞，服务端有应答后会触发回调函数
//            producer.send(record, new Callback() {
//                @Override
//                public void onCompletion(RecordMetadata recordMetadata, Exception e) {
//                    if(null != e){
//                        System.out.println("消息发送失败,"+e.getMessage());
//                        e.printStackTrace();
//                    }else{
//                        String topic = recordMetadata.topic();
//                        long offset = recordMetadata.offset();
//                        String message = recordMetadata.toString();
//                        System.out.println("message:["+ message+"] sended with topic:"+topic+";offset:"+offset);
//                    }
//                    latch.countDown();
//                }
//            });
        }
        //消息处理完才停止发送者。
//        latch.await();
        producer.close();
    }
}
